package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
import org.apache.hadoop.hbase.mapreduce.SimpleTotalOrderPartitioner;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class Demo7HFileToHBase {

    public static class HFileToHBaseMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue> {

        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split(",");

            String id = split[0];
            ImmutableBytesWritable k = new ImmutableBytesWritable(id.getBytes());


            KeyValue name = new KeyValue(id.getBytes(), "info".getBytes(), "name".getBytes(), split[1].getBytes());
            KeyValue age = new KeyValue(id.getBytes(), "info".getBytes(), "age".getBytes(), split[2].getBytes());
            KeyValue gender = new KeyValue(id.getBytes(), "info".getBytes(), "gender".getBytes(), split[3].getBytes());
            KeyValue clazz = new KeyValue(id.getBytes(), "info".getBytes(), "clazz".getBytes(), split[4].getBytes());

            context.write(k, name);
            context.write(k, age);
            context.write(k, gender);
            context.write(k, clazz);
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        //指定zk连接地址
        conf.set("hbase.zookeeper.quorum", "node2:2181,node3:2181.node4:2181");

        Job job = new Job(conf, "HFileToHBase");
        job.setJarByClass(Demo7HFileToHBase.class);

        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(KeyValue.class);

        job.setMapperClass(HFileToHBaseMapper.class);

        //hbase  默认实现大的一个reduce
        job.setReducerClass(KeyValueSortReducer.class);

        //默认每个分区一个reduce
        //job.setNumReduceTasks(4);

        //排序
        job.setPartitionerClass(SimpleTotalOrderPartitioner.class);


        HTable table = new HTable(conf, "stu3");
        HFileOutputFormat2.configureIncrementalLoad(job, table);

        FileInputFormat.addInputPath(job, new Path("/data/student"));
        FileOutputFormat.setOutputPath(job, new Path("/data/hfile"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);



    }

}
